package com.sjgs.gis.fsrpc;


import com.google.protobuf.ByteString;
import com.sjgs.gis.fstrans.*;
import com.sjgs.gis.utils.QETag;
import io.grpc.CallCredentials;
import io.grpc.Channel;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.netty.shaded.io.netty.util.ReferenceCountUtil;
import io.grpc.stub.ClientCallStreamObserver;
import io.grpc.stub.ClientResponseObserver;
import io.grpc.stub.StreamObserver;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * GRPC文件下载客户端示例程序
 */
public class FileSystemGrpcClientTest {

    private static final Log logger = LogFactory.getLog(FileSystemGrpcClientTest.class.getName());
    private final static int GRPC_CHUNK_SIZE = 2 * 1024 * 1024;
    private final static int MAX_MSG_SIZE = 5 * 1024 * 1024;
    private final String COMPRESSION = "gzip";
    private final FSTransServiceGrpc.FSTransServiceBlockingStub transServiceBlockingStub;
    private final FSTransServiceGrpc.FSTransServiceStub transServiceStub;

    public FileSystemGrpcClientTest(CallCredentials credentials, Channel channel) {
        this.transServiceBlockingStub = FSTransServiceGrpc.newBlockingStub(channel)
                .withInterceptors(new HeaderClientInterceptor())
                .withCallCredentials(credentials)
                .withMaxInboundMessageSize(MAX_MSG_SIZE);
        this.transServiceStub = FSTransServiceGrpc.newStub(channel)
                .withInterceptors(new HeaderClientInterceptor())
                .withCallCredentials(credentials)
                .withMaxInboundMessageSize(MAX_MSG_SIZE);
    }

    public FileInfoResponse fetchChunkInfo(String fkey) {
        FileInfoRequest request = FileInfoRequest.newBuilder()
                .setFkey(fkey)
                .build();
        return transServiceBlockingStub.fetchFileInfo(request);
    }

    public FetchChunkOfFileResponse fetchChunkFromFile(String context, long off, long len) {
        FetchChunkOfFileRequest request = FetchChunkOfFileRequest.newBuilder()
                .setOffset(off)
                .setLength(len)
                .setContext(context)
                .build();
        return transServiceBlockingStub.fetchChunkFromFile(request);
    }


    /**
     * 异步下载单片
     *
     * @param fkey
     * @return
     */
    public CountDownLatch asyncfetchChunk(String fkey, int chunkTotal) {
        final CountDownLatch finishLatch = new CountDownLatch(chunkTotal);
        //用户文件空间及bucket路径
        String dir = "/test";
        try {
            for (int i = 0; i < chunkTotal; i++) {
                FetchChunkRequest request = FetchChunkRequest.newBuilder()
                        .setFkey(fkey)
                        .setIndex(i)
                        .setFdir(dir)
                        .build();
                transServiceStub.fetchChunk(request, new StreamObserver<FetchChunkResponse>() {
                    @Override
                    public void onNext(FetchChunkResponse fetchChunkResponse) {
                        logger.info("async-" + fetchChunkResponse.getIndex() + ":" + fetchChunkResponse.getChunk().toByteArray().length);
                        ReferenceCountUtil.safeRelease(fetchChunkResponse);
                    }

                    @Override
                    public void onError(Throwable throwable) {
                        finishLatch.countDown();
                    }

                    @Override
                    public void onCompleted() {
                        finishLatch.countDown();
                    }
                });
            }
        } catch (Exception e) {
            throw e;
        }
        return finishLatch;
    }

    /**
     * 同步取单片
     *
     * @param fkey
     * @param index
     * @return
     */
    public FetchChunkResponse fetchChunk(String fkey, int index) {
        //用户文件空间及bucket路径
        String dir = "/test";
        FetchChunkRequest request = FetchChunkRequest.newBuilder()
                .setFkey(fkey)
                .setIndex(index)
                .setFdir(dir)
                .build();
        return transServiceBlockingStub.fetchChunk(request);
    }

    public MakeFileResponse makeFile(String path, String dir, String storeType) throws IOException {
        if (Files.notExists(Paths.get(path)) || !Files.isReadable(Paths.get(path))) {
            return MakeFileResponse.newBuilder().setContext("false").build();
        }
        File file = new File(path);
        long fsize = file.length();
        String filename = file.getName();
        String fkey = QETag.calcETag(path) + "@" + filename.replaceAll("@", "_");
        MakeFileRequest request = MakeFileRequest.newBuilder()
                .setFkey(fkey)
                .setFsize(fsize)
                .setFdir(dir)
                .setStoretype(storeType)
                .build();
        return transServiceBlockingStub.mkFile(request);
    }

    public CountDownLatch putFile(String path, String context) {
        final CountDownLatch done = new CountDownLatch(1);
        if (Files.notExists(Paths.get(path)) || !Files.isReadable(Paths.get(path))) {
            done.countDown();
            return done;
        }
        ClientResponseObserver<PutChunkRequest, PutChunkResponse> responseObserver =
                new ClientResponseObserver<PutChunkRequest, PutChunkResponse>() {
                    ClientCallStreamObserver<PutChunkRequest> requestStream;

                    @Override
                    public void beforeStart(final ClientCallStreamObserver<PutChunkRequest> requestStream) {
                        this.requestStream = requestStream;
                        // @1设置手动流量控制
                        requestStream.disableAutoInboundFlowControl();
                        // @2 当Consumer端有足够空间时自动回调
                        // 序列化protobuf先发送到缓存区（还未到Server端）
                        // Server端需要调用request()向Client拉取消息
                        try {
                            requestStream.setOnReadyHandler(new Runnable() {
                                File file = new File(path);
                                FileInputStream fin = new FileInputStream(file);
                                FileChannel inChannel = fin.getChannel();
                                ByteBuffer byteBuffer = ByteBuffer.allocate(GRPC_CHUNK_SIZE);
                                Long index = 0L;
                                Long offset = 0L;

                                @Override
                                public void run() {
                                    int eof = 0;
                                    while (requestStream.isReady()) {
                                        try {
                                            if ((eof = inChannel.read(byteBuffer)) != -1) {
                                                byteBuffer.flip();
                                                offset += eof;
                                                byte[] bytes = new byte[eof];
                                                byteBuffer.get(bytes);
                                                byteBuffer.clear();
                                                String hash = QETag.calcETag(bytes);
                                                logger.info(index + ":" + hash + ":" + eof);
                                                PutChunkRequest request = PutChunkRequest
                                                        .newBuilder()
                                                        .setContext(context)
                                                        .setHash(hash)
                                                        .setIndex(index)
                                                        .setStart(index * GRPC_CHUNK_SIZE)
                                                        .setEnd(offset)
                                                        .setChunk(ByteString.copyFrom(bytes))
                                                        .build();
                                                // @3 将消息发送到缓存区
                                                requestStream.onNext(request);
                                                index++;
                                            } else {
                                                requestStream.onCompleted();
                                            }
                                        } catch (IOException e) {
                                            e.printStackTrace();
                                        }
                                    }
                                }
                            });
                        } catch (FileNotFoundException e) {
                            e.printStackTrace();
                        }
                    }

                    @Override
                    public void onNext(PutChunkResponse putChunkResponse) {
                        logger.info(putChunkResponse.getChunkId() + " : " + putChunkResponse.getChunkAddr());
                        // @6通知继续发送
                        requestStream.request(1);
                    }

                    @Override
                    public void onError(Throwable throwable) {
                        done.countDown();
                    }

                    @Override
                    public void onCompleted() {
                        done.countDown();
                    }
                };
        transServiceStub.putFile(responseObserver);
        return done;
    }


    /**
     * 模拟取出其它节点的片数据
     *
     * @param args
     */
    public static void main(String[] args) throws InterruptedException {
        String target = "192.168.1.103:8779";
        String fkey = "lmPeXO-fsgLD_GKL3yY870kAVlU8@en_visual_studio_community_2015_x86_dvd_6847364.iso";
        String dir = "/test";
        String path = "E:/迅雷下载/nosqlbooster4mongo-4.7.5.exe";
        String clientId = "default-client";
        Long start = System.currentTimeMillis();
        CallCredentials credentials = new JwtCredential(clientId);
        ManagedChannel channel = ManagedChannelBuilder
                .forTarget(target)
                .usePlaintext()
                .enableRetry()
                .maxRetryAttempts(5)
                .keepAliveTime(5, TimeUnit.MINUTES)
                .keepAliveWithoutCalls(true)
                .keepAliveTimeout(10, TimeUnit.MINUTES)
                .perRpcBufferLimit(MAX_MSG_SIZE)
                .build();
        FileSystemGrpcClientTest client = new FileSystemGrpcClientTest(credentials, channel);

        FileInfoResponse FileInfoResponse = client.fetchChunkInfo(fkey);
        long length = FileInfoResponse.getIndexList().size();
        String context = FileInfoResponse.getContext();
        System.out.println(context);
        for (int i = 0; i < length; i++) {
            long index = FileInfoResponse.getIndex(i);
            long off = FileInfoResponse.getOffset(i);
            long len = FileInfoResponse.getLength(i);
            FetchChunkOfFileResponse fetchChunkOfFileResponse = client.fetchChunkFromFile(context, off, len);
            logger.info("fetchChunkFromFile:" + index + ":" + fetchChunkOfFileResponse.getChunk().toByteArray().length);
            //other
        }
        System.out.println("fetchChunkFromFile cost time :" + String.valueOf((System.currentTimeMillis() - start)) + "ms");

//        try {
//            MakeFileResponse makeFileResponse = client.makeFile(path, dir, "0");
//            context = makeFileResponse.getContext();
//            System.out.println(context);
//            if (!makeFileResponse.getContext().equals("AQEvdGVz")) {
//                logger.info("本地文件不存在或者服务器已存在相同文件");
//            } else {
//                long fsize = new File(path).length();
//                CountDownLatch downLatch = client.putFile(path, context);
//                //2G=>120s
//                long wait_time = Math.round(fsize / GRPC_CHUNK_SIZE);
//                if (!downLatch.await(wait_time, TimeUnit.SECONDS)) {
//                    logger.warn("async fetchChunk can not finish within " + wait_time + " seconds");
//                }
//            }
//            System.out.println("upload file cost time :" + String.valueOf((System.currentTimeMillis() - start)) + "ms");
//        } catch (IOException e) {
//            e.printStackTrace();
//        } finally {
//            channel.shutdownNow().awaitTermination(2, TimeUnit.MINUTES);
//        }

//        start = System.currentTimeMillis();
//        FileInfoResponse = client.fetchChunkInfo(fkey);
//        length = FileInfoResponse.getIndexList().size();
//        for (int i = 0; i < length; i++) {
//            long index = FileInfoResponse.getIndex(i);
//            long off = FileInfoResponse.getOffset(i);
//            long len = FileInfoResponse.getLength(i);
//            FetchChunkOfFileResponse fetchChunkOfFileResponse = client.fetchChunkFromFile(fkey, off, len);
//            logger.info("fetchChunkFromFile:" + index + ":" + fetchChunkOfFileResponse.getChunk().toByteArray().length);
//            //write chunk
//        }
//        System.out.println("fetchChunkFromFile cost time :" + String.valueOf((System.currentTimeMillis() - start)) + "ms");

    }
}
